package com.gitee.limaozhi.socket.lesson03.example3;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 解码处理器
 *
 * @author hzlimaozhi
 * @date 2019/08/07 11:19
 */
public class DecoderHandler extends HandlerAdapter {

    /** 上次处理器发来的请求数 */
    private int prevReq = -1;
    /** 接收的请求数 */
    private AtomicInteger recvReq = new AtomicInteger(0);
    /** 发往下个处理的请求数 */
    private AtomicInteger req = new AtomicInteger(0);
    /** 是否已触发结束 */
    private boolean fireFinish = false;

    private static final char SEPARATOR = '\n';
    private LinkedBlockingQueue<byte[]> queue = new LinkedBlockingQueue<>();
    private ReentrantLock lock = new ReentrantLock();
    private volatile byte[] left = null;

    @Override
    public Handler handle(Object object) {
        context().execute(() -> {
            try {
                queue.put((byte[]) object);
            } catch (InterruptedException e) {
                e.printStackTrace();
                return;
            }

            if (lock.tryLock()) {
                try {
                    while (true) {
                        byte[] item = queue.poll();
                        if (item == null) {
                            break;
                        }
                        handleItem(item);
                        // req增加在recvReq增加在前，避免在req停止增加前发送req，导致提前shutdownOutput
                        recvReq.incrementAndGet();
                        fireFinish();
                    }
                } finally {
                    lock.unlock();
//                    context().print(context().getName() + "：Decoder：线程退出");
                }
            }
        });
        return this;
    }

    private void handleItem(byte[] bytes) {
        int start = 0;
        for (int i = 0; i < bytes.length; i++) {
            if (bytes[i] == SEPARATOR) {
                int leftLength = left != null ? left.length : 0;
                int iterLength = i - start;
                int length = leftLength + iterLength;
                if (length > 0) {
                    byte[] item = new byte[length];
                    if (leftLength > 0) {
                        System.arraycopy(left, 0, item, 0, leftLength);
                    }
                    if (iterLength > 0) {
                        System.arraycopy(bytes, start, item, leftLength, iterLength);
                    }
                    // 发往下个处理器
                    next().handle(new String(item));
                    req.incrementAndGet();
                }
                left = null;
                // 下一个的起点
                start = i + 1;
            }
        }

        {
            int length = bytes.length - start;
            if (length > 0) {
                left = new byte[length];
                System.arraycopy(bytes, start, left, 0, length);
            }
        }
    }

    @Override
    public Handler finish(int prevReq) {
        this.prevReq = prevReq;
        // 如果finish比较晚，则在此触发
        fireFinish();
        return this;
    }

    private synchronized void fireFinish() {
        if (!fireFinish) {
            if (prevReq == recvReq.get()) {
                fireFinish = true;
//                context().print(context().getName() + "：Decoder：" + req.get());
                next().finish(req.get());
            }
        }
    }

}
